package com.thiswhyme.republish.rabbitmq.core;

import com.thiswhyme.republish.rabbitmq.core.domain.MessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;

import com.thiswhyme.republish.common.UidGenService;
import com.thiswhyme.republish.rabbitmq.config.BindingPlusProperties;
import com.thiswhyme.republish.rabbitmq.config.BindingServicePlusProperties;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishMqHeaders;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishStatus;

/**
 * @description: 全局消息拦截器-用于处理重发消息
 * @author: thiswhyme
 * @date: 2021/2/5
 */
public class RepublishChannelInterceptor implements ChannelInterceptor {

    public static final Logger log = LoggerFactory.getLogger(RepublishChannelInterceptor.class);

    private CompositeMessageConverter compositeMessageConverter;
    private RepublishRedisOperation republishRedisOperation;
    private BindingServicePlusProperties bindingServicePlusProperties;
    private MessageInfoService messageInfoService;
    private UidGenService uidGenService;

    public RepublishChannelInterceptor(CompositeMessageConverter compositeMessageConverter,
                                       RepublishRedisOperation republishRedisOperation,
                                       MessageInfoService messageInfoService,
                                       BindingServicePlusProperties bindingServicePlusProperties,
                                       UidGenService uidGenService) {
        this.compositeMessageConverter = compositeMessageConverter;
        this.republishRedisOperation = republishRedisOperation;
        this.messageInfoService = messageInfoService;
        this.bindingServicePlusProperties = bindingServicePlusProperties;
        this.uidGenService = uidGenService;
    }

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        return doBizPreSend(message, channel);
    }

    /**
     * 自定义消息预发送处理
     * @param message
     * @param channel
     * @return
     */
    private Message<?> doBizPreSend(Message<?> message, MessageChannel channel) {
        try {
            if(null != bindingServicePlusProperties.getBindings()) {
                BindingPlusProperties bindingPlusProperties = bindingServicePlusProperties.getBindings().get(channel.toString());
                if(null != bindingPlusProperties) {
                    if(null != message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE)) {
                        log.debug("【重发拦截器】-input类型通道的消息，不处理，通道名：{}，消息UUID：{}", channel.toString(), message.getHeaders().getId().toString());
                        return message;
                    }

                    if(bindingPlusProperties.getEnableRepublish()) {
                        //开启重试
                        message = doIfRepublishEnabled(message, channel, bindingPlusProperties);
                    } else {
                        //未开启重试
                        message = doIfRepublishUnEnabled(message, channel, bindingPlusProperties.getEnableStore());
                    }
                }
            }
            log.debug("【重发拦截器】-消息发送预处理正常结束，通道名：{}，消息UUID：{}", channel.toString(), message.getHeaders().getId().toString());
        } catch (Exception e) {
            log.error("【重发拦截器】-消息发送预处理异常结束，通道名：{}，消息UUID：{}", channel.toString(), message.getHeaders().getId().toString(), e);
        }
        return message;
    }

    /**
     * 开启重试处理
     * @param message
     * @param channel
     * @param bindingPlusProperties
     * @return
     */
    private Message<?> doIfRepublishEnabled(Message<?> message, MessageChannel channel, BindingPlusProperties bindingPlusProperties) {
        Object current = message.getHeaders().get(PublishMqHeaders.HEADER_SEND_RETRY_TIMES);

        //需要重试的消息，第一次发送时做如下处理：消息内容入库，消息置入redis待ack的zset
        if(null == current) {
            Object content = compositeMessageConverter.fromMessage(message, bindingPlusProperties.getMessageContentClazz());
            Long uniqueId = uidGenService.fetchUid();
            message = MessageBuilder.fromMessage(message)
                    .setHeader(PublishMqHeaders.HEADER_RETRY_UNIQUE_ID, uniqueId)
                    .setHeader(PublishMqHeaders.HEADER_SEND_RETRY_TIMES, 0)
                    .build();
            MessageInfo msg = messageInfoService.recordMessage(uniqueId, channel.toString(),
                    bindingPlusProperties.getMessageContentClazz().getName(), message.getHeaders(), content,
                    PublishStatus.SEND_TO_ACK, MessageType.PRODUCE);
            republishRedisOperation.addToAckZSet( msg.getId(), System.currentTimeMillis()
                    + bindingServicePlusProperties.getAckMillis());
        }
        return message;
    }

    /**
     * 未开启重试处理
     * @param message
     * @param channel
     * @param enableStore
     * @return
     */
    private Message<?> doIfRepublishUnEnabled(Message<?> message, MessageChannel channel, Boolean enableStore) {
        if(enableStore) {
            Object content = compositeMessageConverter.fromMessage(message, String.class);
            messageInfoService.recordMessage(null, channel.toString(), String.class.getName(),
                    message.getHeaders(), content, PublishStatus.NONE, MessageType.PRODUCE);
        } else {
            log.debug("【重发拦截器】-消息无需重发无需入库，通道名：{}", channel.toString());
        }
        return MessageBuilder.fromMessage(message).build();
    }
}
